package com.apps.sdses.flink141.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/6/15 0:18
 */
public class Kafka2Print_Geom {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "  geom_data STRING,\n" +
                "  car_nos ARRAY<STRING>,\n" +
                "  msg_id BIGINT,\n" +
                "  event_time TIMESTAMP(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'test_topic',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'properties.group.id' = 'my_group_id',\n" +
                "  'format' = 'json',\n" +
                "  'scan.startup.mode' = 'latest-offset'," +
                "  'json.fail-on-missing-field' = 'false',\n" +
                "  'json.ignore-parse-errors' = 'true',\n" +
                "  'json.timestamp-format.standard' = 'ISO-8601'\n" +
                ")\n");

        tableEnv.executeSql("CREATE TABLE print_table WITH ('connector' = 'print')\n" +
                "LIKE sourceTable (EXCLUDING ALL)");

        tableEnv.executeSql("insert into print_table select * from sourceTable");




    }
}
